package com.atguigu;

import com.streamxhub.streamx.flink.core.StreamEnvConfig;
import com.streamxhub.streamx.flink.core.java.function.SQLFromFunction;
import com.streamxhub.streamx.flink.core.java.sink.JdbcSink;
import com.streamxhub.streamx.flink.core.java.source.KafkaSource;
import com.streamxhub.streamx.flink.core.scala.StreamingContext;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;


/**
 * @Author lzc
 * @Date 2022/4/27 15:20
 */
public class Streamx_mysql {
    public static void main(String[] args) {
        StreamEnvConfig config = new StreamEnvConfig(args, null);
        StreamingContext ctx = new StreamingContext(config);
    
        SingleOutputStreamOperator<Tm> stream1 = new KafkaSource<String>(ctx)
            .topic("s1")
            .getDataStream()
            .map(r -> {
                String[] data = r.value().split(",");
                return new Tm(Long.valueOf(data[0]), data[1], data[2]);
            });
        
            
        
        
      new JdbcSink<Tm>(ctx)
            .sql(new SQLFromFunction<Tm>() {
                @Override
                public String from(Tm tm) {  // sql语句
                    return String
                        .format("insert into base_trademark values(%d, '%s', '%s')", tm.getId(), tm.getTm_name(), tm.getLogo_url());
                }
            })
            .sink(stream1);
    
    
    
    
        ctx.start();
    }
}
